package com.aiobc.mq_transaction;


import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 *
 * product生产端监听发送的消息，根据本地事务返回commit还是rollback，Borker回查本地事务并执行本地事务
 * 事务监听器，重写执行本地事务方法以及事务回查方法
 *
 * 事务消息共有三种状态，提交状态、回滚状态、中间状态：
 * TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
 * TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
 * TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。
 *
 */

public  class TransactionListenerImpl implements TransactionListener  {

    /**
     * executeLocalTransaction：执行本地事务；
     * checkLocalTransaction：回查本地事务状态，根据这次回查的结果来决定此次事务是提交还是回滚；
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        String msgKey = msg.getKeys();
        switch (msgKey) {
            case "Num0":
            case "Num1":
                // 明确回复回滚操作，消息将会被删除，不允许被消费。
                System.out.println("回滚===="+msgKey);
                return LocalTransactionState.ROLLBACK_MESSAGE;
            case "Num8":
            case "Num9":
                // 消息无响应,代表需要回查本地事务状态来决定是提交还是回滚事务
                System.out.println("无响应===="+msgKey);
                return LocalTransactionState.UNKNOW;
            default:
                // 消息通过，允许消费者消费消息
                System.out.println("通过===="+msgKey);
                return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("回查本地事务状态,消息Key: " + msg.getKeys() + ",消息内容: " + new String(msg.getBody()));

        // 需要根据业务，查询本地事务是否执行成功，这里直接返回COMMIT
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
